package cn.ai.boot.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

/**
 * @author jie.zhao@dms365.com
 * @Date 2025/7/2 11:20
 */
public class AsyncService {
    private static final Logger LOGGER = LoggerFactory.getLogger(AsyncService.class);


    public <T> void executeConcurrentTasks(ExecutorService threadPoolService,
                                           Map<String, Map<String, Object>> map,
                                           int maxConcurrency,
                                           Consumer<String> taskConsumer, String threadPoolName) {

        if (map.isEmpty()) {
            return;
        }
        LOGGER.info("executeConcurrentTasks.threadPoolName.start:{}", threadPoolName);
        Semaphore semaphore = new Semaphore(maxConcurrency);
        AtomicBoolean isException = new AtomicBoolean(false);
        List<Future<Boolean>> futures = new ArrayList<>();

        for (String key : map.keySet()) {
            if (isException.get()) {
                throw new RuntimeException("queryTransformUpdateAsync.isException.true");
            }
            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                isException.set(true);
                LOGGER.error("获取许可异常", e);
                throw new RuntimeException("saveFolderAsync.InterruptedException", e);
            }
            futures.add(threadPoolService.submit(() -> {
                try {
                    taskConsumer.accept(key);
                    return true;
                } catch (Exception e) {
                    isException.set(true);
                    LOGGER.error("Async task failed", e);
                    throw e;
                } finally {
                    semaphore.release();
                }
            }));
        }
        LOGGER.info("executeConcurrentTasks：{}", futures.size());
        for (Future<Boolean> future : futures) {
            try {
                future.get();
            } catch (Exception e) {
                throw new RuntimeException("saveFolderAsync.InterruptedException", e);
            }
        }
        if (isException.get()) {
            throw new RuntimeException("Async execution failed");
        }
        LOGGER.info("executeConcurrentTasks.threadPoolName.end:{}", threadPoolName);
    }


    public <T> void executeConcurrentTasks(ExecutorService threadPoolService,
                                           Collection<List<T>> list,
                                           int maxConcurrency,
                                           Consumer<List<T>> taskConsumer,
                                           String threadPoolName) {

        if (list.isEmpty()) {
            return;
        }
        LOGGER.info("executeConcurrentTasks.threadPoolName.start:{}", threadPoolName);
        Semaphore semaphore = new Semaphore(maxConcurrency);
        AtomicBoolean isException = new AtomicBoolean(false);
        List<Future<Boolean>> futures = new ArrayList<>();
        for (List<T> ts : list) {
            if (isException.get()) {
                throw new RuntimeException("queryTransformUpdateAsync.isException.true");
            }
            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                isException.set(true);
                LOGGER.error("获取许可异常", e);
                throw new RuntimeException("saveFolderAsync.InterruptedException", e);
            }
            futures.add(threadPoolService.submit(() -> {
                try {
                    taskConsumer.accept(ts);
                    return true;
                } catch (Exception e) {
                    isException.set(true);
                    LOGGER.error("Async task failed", e);
                    throw e;
                } finally {
                    semaphore.release();
                }
            }));
        }
        LOGGER.info("executeConcurrentTasks：{}", futures.size());
        for (Future<Boolean> future : futures) {
            try {
                future.get();
            } catch (Exception e) {
                throw new RuntimeException("saveFolderAsync.InterruptedException", e);
            }
        }
        if (isException.get()) {
            throw new RuntimeException("Async execution failed");
        }
        LOGGER.info("executeConcurrentTasks.threadPoolName.end:{}", threadPoolName);
    }
}
